/*
    Copyright 2010 Barnet Wagman

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

        http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
 */

package com.norbl.util.sdb;

import com.xerox.amazonws.sdb.*;
import java.util.*;
import com.norbl.util.*;
/**
 *
 * @author Barnet Wagman
 */
public class SdbOps {

    public static String ITEM_NAME = "itemName";
    public static int MAX_ATTRIBUTE_LEN = 1024 - 1;

    public static int MAX_PUT_BUFFER_SIZE = 32;
    public static long MAX_PUT_FLUSH_INTERVAL = 1000L * 20L;

    long retryIntervalMillis;
    long maxWaitMillis;

    private String awsAccessID;
    private String awsSecretKey;

    private SimpleDB sdb;

    private HashMap<String,Domain> domainHt;

    PutThread putThread;

    public SdbOps(String awsAccessID, String awsSecretKey,
                  long retryIntervalMillis, long maxWaitMillis) {
        this.awsAccessID = awsAccessID;
        this.awsSecretKey = awsSecretKey;
        this.retryIntervalMillis = retryIntervalMillis;
        this.maxWaitMillis = maxWaitMillis;
        sdb = new SimpleDB(awsAccessID, awsSecretKey, true);
        domainHt = new HashMap<String, Domain>();
    }

    public SdbOps cleanClone() {
        return( new SdbOps(awsAccessID,awsSecretKey,
                           retryIntervalMillis,maxWaitMillis) );
    }

        // -------- Public ops ----------------

        /** This method always queries SDB. It does not use
         *  the internal domain hashtable.
         *
         * @param domainName
         * @return true if it exits, else false.
         */
    public boolean domainExist(String domainName) throws SDBException {

        testDomainName(domainName);
        
        ListDomainsResult r = sdb.listDomains();
        for ( Domain d : r.getDomainList() ) {
            if ( domainName.equals(d.getName()) ) return(true);
        }
        return(false);
    }

        /** If the domain already exist, this method has no
         *  effect and does not throw and exception.
         *
         * @param domainName
         */
    public void createDomain(String domainName) throws SDBException {
        
        testDomainName(domainName);

        Domain d = sdb.createDomain(domainName);
        if ( !domainHt.containsKey(domainName) ) {
            domainHt.put(domainName,d);
        }
    }

    public void deleteDomain(String domainName) throws SDBException {
        Domain d = getDomain(domainName);
        sdb.deleteDomain(d);        
    }

        /** Creates or sets the attributes of an item. Note that
         *  SDB does not allow the creation of an empty item,
         *  so <tt>attributes</tt> cannot be empty.<p>
         *
         *  The item name uniquely identifes an item, so
         *  if the specified item exists, its attribute values
         *  will be replaced (if it has values for the specified
         *  attributes).<p>
         *
         *  If any of the specified attributes already exists,
         *  those attribute values will be replaced by this operation.
         *
         * @param domainName
         * @param itemName
         * @param attributes
         * @param retry if true and the put fails, this method
         *      retries a {@link #retryIntervalMillis} until
         *      {@link #maxWaitMillis} is reached.
         */
    public void putItem(String domainName,
                        String itemName,
                        PropertyList attributes,
                        boolean retry) throws SDBException {
        if ( retry ) putItem(domainName,itemName,attributes,
                             retryIntervalMillis, maxWaitMillis);
        else putItem(domainName,itemName,attributes);
    }


    private void putItem(String domainName,
                        String itemName,
                        PropertyList attributes) throws SDBException {
        
        if ( (attributes == null) || (attributes.size() < 1) )
            throw new NoAttributesException("The attribute list was empty or null.");

        Domain d = getDomain(domainName);
        
        Item item = d.getItem(truncToSize(itemName));
        item.putAttributes(createAttributeList(attributes));
    }

    public void putItemBackground(String domainName,
                                  String itemName,
                                  PropertyList attributes) {
        if ( putThread == null ) startPutThread();
        putThread.putPi(domainName, itemName, attributes);
    }

    private void putItem(String domainName,
                        String itemName,
                        PropertyList attributes,
                        long retryIntervalMillis,
                        long maxWaitMillis) throws SDBException {

        long t0 = System.currentTimeMillis();
        SDBException sx;
        do {
            try {
                putItem(domainName,itemName,attributes);
                return;
            }
            catch(SDBException x) {
                sx = x;
                Time.nap(retryIntervalMillis);
            }
        }
        while ( !Time.timedOut(t0,maxWaitMillis) );

        throw sx;        
    }

        /** Gets all items in a domain.  Each item is a {@link PropertyList}
         *  with a {@link Property} for each attribute and the item
         *  name in a <tt>Property</tt> named "itemName".         
         *
         * @param domainName
         * @param retry if true and a get fails, this method
         *      retries a {@link #retryIntervalMillis} until
         *      {@link #maxWaitMillis} is reached.
         * @return contents of items in the domain.
         */
    public List<PropertyList> getItems(String domainName, boolean retry)
        throws SDBException {

        if ( retry )
            return( getItems(domainName,
                             retryIntervalMillis,
                             maxWaitMillis) );
        else return( getItems(domainName) );
    }

        /** Gets all items in a domain.  Each item is a {@link PropertyList}
         *  with a {@link Property} for each attribute and the item
         *  name in a <tt>Property</tt> named "itemName".
         *
         * @param domainName
         * @param retry if true and a get fails, this method
         *      retries a {@link #retryIntervalMillis} until
         *      {@link #maxWaitMillis} is reached.
         * @return contents of items in the domain that match the query.
         */
    public List<PropertyList> getItems(String domainName, String query,
                                      boolean retry)
        throws SDBException {

        if ( retry )
            return( getItems(domainName,query,
                             retryIntervalMillis,
                             maxWaitMillis) );
        else return( getItems(domainName,query) );
    }

    private List<PropertyList> getItems(String domainName)
        throws SDBException {
        String qs = "select * from " + domainName;
        return(getItems(domainName,qs));
    }

    private List<PropertyList> getItems(String domainName,
                                       String query)
        throws SDBException {

        Domain d = getDomain(domainName);

        List<SdbItem> items = getItems(d,query);

        List<PropertyList> pls = new ArrayList<PropertyList>();

        for ( SdbItem item : items ) {
            pls.add(item.toPropertyList());
        }

        return(pls);
    }

    private List<PropertyList> getItems(String domainName,
                                       String query,
                                       long retryIntervalMillis,
                                       long maxWaitMillis)
        throws SDBException {

        long t0 = System.currentTimeMillis();
        SDBException sx;
        do {
            try {
                return(getItems(domainName,query));
            }
            catch(SDBException x) {
                sx = x;
                Time.nap(retryIntervalMillis);
            }
        }
        while ( !Time.timedOut(t0,maxWaitMillis) );
        throw sx;
    }

    private List<PropertyList> getItems(String domainName,
                                       long retryIntervalMillis,
                                       long maxWaitMillis)
        throws SDBException {
        String qs = "select * from " + domainName;
        return(getItems(domainName,qs,
                        retryIntervalMillis,maxWaitMillis));
    }

    public boolean deleteItem(String domainName,String itemName,
                             boolean retry)
        throws SDBException {
        if ( retry ) return(deleteItem(domainName,itemName,
                                       retryIntervalMillis,
                                       maxWaitMillis));
        else return(deleteItem(domainName,itemName));
    }

    public void deleteItems(String domainName, String query,boolean retry)
        throws SDBException {
        if ( retry ) deleteItems(domainName,query,
                                 retryIntervalMillis,
                                 maxWaitMillis);
        else deleteItems(domainName,query);
    }

    public void deleteItems(String domainName,boolean retry)
        throws SDBException {

        String q = "select * from " + domainName;

        if ( retry ) deleteItems(domainName,q,
                                 retryIntervalMillis,
                                 maxWaitMillis);
        else deleteItems(domainName,q);
    }


    private boolean deleteItem(String domainName,String itemName)
        throws SDBException {

        Domain d = getDomain(domainName);
        if ( d != null ) {
            d.deleteItem(truncToSize(itemName));
            return(true);
        }
        else return(false);
    }

    private boolean deleteItem(String domainName, String itemName,
                              long retryIntervalMillis,
                              long maxWaitMillis)
        throws SDBException {

        long t0 = System.currentTimeMillis();
        SDBException sx;
        do {
            try {
                return(deleteItem(domainName,itemName));
            }
            catch(SDBException x) {
                sx = x;
                Time.nap(retryIntervalMillis);
            }
        }
        while ( !Time.timedOut(t0,maxWaitMillis) );
        throw sx;
    }

    private void deleteItems(String domainName, String query)
        throws SDBException {

        Domain d = getDomain(domainName);
        if ( d == null ) return;

        List<SdbItem> items = getItems(d,query);

        for ( SdbItem item : items ) {
            d.deleteItem(item.itemName);
        }
    }

    private void deleteItems(String domainName)
        throws SDBException {
        deleteItems(domainName,"select * from " + domainName);
    }

    private void deleteItems(String domainName, String query,
                            long retryIntervalMillis,
                            long maxWaitMillis)
        throws SDBException {

        long t0 = System.currentTimeMillis();
        SDBException sx;
        do {
            try {
                deleteItems(domainName,query);
                return;
            }
            catch(SDBException x) {
                sx = x;
                Time.nap(retryIntervalMillis);
            }
        }
        while ( !Time.timedOut(t0,maxWaitMillis) );
        throw sx;
    }
        // ------------------------------------------

    private List<SdbItem> getItems(Domain d, String query) throws SDBException {

        String nextToken = "";
        List<SdbItem> items = new ArrayList<SdbItem>();

        do {
            try {
                QueryWithAttributesResult r = d.selectItems(query, nextToken);
                Map<String,List<ItemAttribute>> imap = r.getItems();
                for (Iterator<String> it = imap.keySet().iterator();
                     it.hasNext(); ) {
                     String inm = it.next();
                     items.add(new SdbItem(inm,imap.get(inm)));
                }
                nextToken = r.getNextToken();
            }
            catch(SDBException sx) {
                if ( isNoSuchDomainException(sx) ) {
                    return(new ArrayList<SdbItem>());
                }
                else throw sx;
            }
        }
        while (nextToken != null && !nextToken.trim().equals(""));

        return(items);
    }

    boolean isNoSuchDomainException(SDBException x) {
        return(x.toString().contains("specified domain does not exist"));
    }

    List<ItemAttribute> createAttributeList(PropertyList pl) {

        List<ItemAttribute> al = new ArrayList<ItemAttribute>();
        for ( Property p : pl.getProperties() ) {
            if ( p.value != null)
                al.add(new ItemAttribute(p.name,
                                         truncToSize(p.valueToString()),
                                         true));          
        }
        return(al);
    }

    static String truncToSize(String s) {
        if ( s.length() <= MAX_ATTRIBUTE_LEN ) return(s);
        else return( s.substring(0,MAX_ATTRIBUTE_LEN) );
    }

    void testDomainName(String domainName) {
        if ( !isLegalDomainName(domainName) ) {
            throw new IllegalDomainNameException(domainName +
                       " is not a legal domain name.");
        }
    }

        /** WARNING: this method tests domain names per the AWS
         *  SQS documentation, which states that '-' is legal.
         *  However, embedded '-' cause
         *  SDB Tool to fail on queries.  It is not clear where
         *  the bug lies.  To be on the safe side, do not use '-'
         *  in domain names.  '_' does work.
         *
         * @param name
         * @return
         */
    boolean isLegalDomainName(String name) {

        if ( (name == null) || (name.length() < 3) ||
             (name.length() > 255) ) return(false);
        for ( int i = 0; i < name.length(); i++ ) {
            if ( !okChar(name.charAt(i)) ) return(false);
        }
        return(true);
    }

    Domain getDomain(String domainName) throws SDBException {

        testDomainName(domainName);

        Domain d = domainHt.get(domainName);
        if ( d == null ) {
            d = sdb.getDomain(domainName);
            if ( d != null ) domainHt.put(domainName, d);
        }
        return(d);
    }

        /** We need this rigamarole because SDB is very restrictive
         *  re domain names.
         * 
         * @param c
         * @return
         */
    boolean okChar(char c) {
        return( inClosedInterval('a','z',c) ||
                inClosedInterval('A','Z',c) ||
                inClosedInterval('0','9', c) ||
                (c == '_') || (c == '-') || (c == '.') );
    }

    boolean inClosedInterval(char lower, char upper, char c) {
        return( (c >= lower) && (c <= upper) );
    }

        /** This list contains all attributes in the item and
         *  the item name. Note that all values are strings.
         *
         * @param item
         * @return
         * @throws SDBException
         */
    PropertyList toPropertyList(Item item) throws SDBException {

        PropertyList pl = new PropertyList();
        pl.add(ITEM_NAME,item.getIdentifier());

        for ( ItemAttribute ia : item.getAttributes() ) {
            pl.add(ia.getName(),ia.getValue());
        }
        return(pl);
    }

        /** This crap is necessary because <tt>Domain.listItems()</tt>
         *  is obsolete (and fails with the current syntax) and
         *  it's replacement, <tt>Domain.selectItems()</tt>
         *  doesn't return items, just pieces.
         */
    class SdbItem {

        String itemName;
        List<ItemAttribute> iatts;

        SdbItem(String itemName,List<ItemAttribute> iatts) {
            this.itemName = itemName;
            this.iatts = iatts;
        }

        PropertyList toPropertyList() {
            PropertyList pl = new PropertyList();
            pl.add(ITEM_NAME,itemName);
            for ( ItemAttribute ia : iatts ) {
                pl.add(ia.getName(),ia.getValue());
            }            
            return(pl);
        }

    }

        // ------------------------------------------------------

    public void startPutThread(long retryIntervalMillis, long maxWaitMillis) {
        if ( putThread != null ) return;
        putThread = new PutThread();
        Thread t = new Thread(putThread);
        t.start();

            // Wait for the list to exist
        while ( putThread.items == null ) {
            Time.nap(100L);
        }
    }

    public void startPutThread() {
        startPutThread(-1L,-1L);
    }

    public void stopPutThread() {
        if ( putThread == null ) return;
        else putThread.keepRunning = false;
    }

        /** Blocks until the put queue is empty.
         *
         */
    public void waitForBackgroundPuts() {
        if ( putThread == null ) return;
        while ( putThread.items.size() > 0 ) {
            Time.nap(1000L);
        }
    }

    class PutItem {
        String domainName;
        String itemName;
        PropertyList attributes;

        PutItem(String domainName, String itemName, PropertyList attributes) {
            this.domainName = domainName;
            this.itemName = itemName;
            this.attributes = attributes;
        }
    }

    class PutThread implements Runnable {

        boolean keepRunning;
        List<PutItem> items;
        long retryIntervalMillis, maxWaitMillis;
        long lastFlush;

        PutThread(long retryIntervalMillis, long maxWaitMillis) {
            this.retryIntervalMillis = retryIntervalMillis;
            this.maxWaitMillis = maxWaitMillis;
            items =  Collections.synchronizedList(new ArrayList<PutItem>());
        }

        PutThread() {
           this(-1L,-1L);
        }

        void putPi(String domainName,
                   String itemName, PropertyList attributes) {
            if ( (items.size() > MAX_PUT_BUFFER_SIZE) ||
                 ( (System.currentTimeMillis() - lastFlush) > 
                                            MAX_PUT_FLUSH_INTERVAL) ) {                
                flush();
                lastFlush = System.currentTimeMillis();               
            }
            else {               
                items.add( new PutItem(domainName, itemName, attributes) );
            }
        }

        public void run() {

            Thread.currentThread().setPriority(Thread.MIN_PRIORITY);
            keepRunning = true;

            if ( retryIntervalMillis > 1L ) runWithRetries();
            else runWithoutRetries();
        }

        void runWithoutRetries() {
            while (keepRunning) {               
                if ( items.size() > 0 ) {
                    PutItem pi = items.remove(0);
                    try {
                        putItem(pi.domainName,pi.itemName,pi.attributes);                       
                    }
                    catch(Throwable x) {
                        System.out.println(StringUtil.throwableToString(x));
                    }
                    Thread.yield();
                }
                else Time.nap(1000L);
            }
        }

        void runWithRetries() {
            while (keepRunning) {
                synchronized (items) {
                    if ( items.size() > 0 ) {
                        PutItem pi = items.remove(0);
                        try {
                            putItem(pi.domainName,pi.itemName,pi.attributes,
                                    retryIntervalMillis,maxWaitMillis);                           
                        }
                        catch(Throwable x) {
                            System.out.println(StringUtil.throwableToString(x));
                        }
                        Thread.yield();
                    }
                    else Time.nap(1000L);
                }
            }
        }

        void flush() {
            synchronized(items) {
                for ( PutItem pi : items ) {
                    try {
                        if ( retryIntervalMillis > 1L )
                            putItem(pi.domainName,pi.itemName,pi.attributes,
                                        retryIntervalMillis,maxWaitMillis);
                        else putItem(pi.domainName,pi.itemName,pi.attributes);
                    }
                    catch(Throwable x) {
                        System.out.println(StringUtil.throwableToString(x));
                    }
                }
                items.clear();
            }
        }
    }
}
